package ru.ifmo.croak.impl.replication.team1.sync;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.ifmo.croak.api.Message;
import ru.ifmo.croak.api.MessageQuery;
import ru.ifmo.croak.api.QueryRequest;
import ru.ifmo.croak.impl.replication.team1.LocalContext;
import ru.ifmo.croak.impl.replication.team1.Pair;
import ru.ifmo.croak.impl.replication.team1.io.MessageInputStream;
import ru.ifmo.croak.impl.replication.team1.io.MessageOutputStream;

import java.io.IOException;
import java.net.Socket;
import java.util.Iterator;
import java.util.List;
import java.util.UUID;

public class SynchronizerWorker implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(SynchronizerWorker.class);
    private static final int MESSAGES_PER_ROUND = 1000;

    private final Socket socket;
    private final LocalContext localContext;
    private final boolean accept;

    public SynchronizerWorker(Socket socket, boolean accept, LocalContext localContext) {
        this.socket = socket;
        this.localContext = localContext;
        this.accept = accept;
    }

    public void run() {
        try {
            log.info("start synchronizing with {}", socket);
            log.trace("creating streams ");
            MessageOutputStream messageOutputStream = new MessageOutputStream(socket.getOutputStream());
            MessageInputStream messageInputStream = new MessageInputStream(socket.getInputStream());
            log.trace("created streams ");
            log.trace(" accept = {}", accept);
            if (accept) {
                messageOutputStream.writeBoolean(true);
                messageOutputStream.flush();
                if (messageInputStream.nextBoolean()) {
                    process(messageOutputStream, messageInputStream);
                } else {
                    log.info("Other side refuse synchronization {}", socket);
                }
            } else {
                messageOutputStream.writeBoolean(false);
                messageOutputStream.flush();
                log.info("We refuse synchronization {}", socket);
                messageInputStream.nextBoolean();
            }
            messageOutputStream.close();
            messageInputStream.close();
        } catch (IOException e) {
            log.warn("Unexpected IOException, maybe network goes down?", e);
        } finally {
            try {
                socket.close();
            } catch (IOException e) {
                log.warn("Unexpected IOException, while closing", e);
            }
        }
    }

    private void process(MessageOutputStream messageOutputStream, MessageInputStream messageInputStream) throws IOException {
        QueryRequest queryRequest = new QueryRequest();
        queryRequest.setLimit(Integer.MAX_VALUE);
        MessageQuery messageQuery = localContext.getQueryService().createQuery(queryRequest);
        List<Message> messageList = messageQuery.getMessages();
        Iterator<Message> messageIterator = messageList.iterator();

        int mySize = messageList.size();
        messageOutputStream.writeInt(mySize);
        messageOutputStream.flush();
        log.trace("written size {}", mySize);
        int hisSize = messageInputStream.nextInt();
        log.debug("ready for getting {} messages and sending {} messages", hisSize, mySize);

        Pair<Long, UUID> pair = localContext.getLocalTimeAndXor();
        Long deadLine = pair.getFirst();
        UUID xor = pair.getSecond();
        log.trace("my xor = {}, deadLine = {}", xor, deadLine);
        UUID theirXor = null;
        while (true) {
            log.info("remain getting {} messages and sending {} messages", hisSize, mySize);
            log.trace("xor = {}, theirXor = {}", xor, theirXor);
            messageOutputStream.writeUUID(xor);
            messageOutputStream.flush();
            theirXor = messageInputStream.nextUUID();
            if (xor.equals(theirXor)) {
                log.debug("xors equals! breaking");
                break;
            }
            if (!messageIterator.hasNext() && hisSize == 0) {
                log.warn("Synchronization unsuccessfully with {} :(", socket);
                messageQuery.close();
                return;
            }

            for (int i = 0; i < MESSAGES_PER_ROUND && messageIterator.hasNext(); i++) {
                Message message = messageIterator.next();
                mySize--;
                messageOutputStream.writeMessage(message);
                log.trace("sent {}", message);
            }
            messageOutputStream.flush();

            for (int i = 0; i < MESSAGES_PER_ROUND && hisSize > 0; i++) {
                Message message = messageInputStream.nextMessage();
                Long timeOfArrival = localContext.getLocalTimeOfArrival(message.getId());
                log.trace("timeOfArrival = {}, deadline = {}", timeOfArrival, deadLine);
                if (timeOfArrival == null || timeOfArrival > deadLine) {
                    xor = LocalContext.xorUUID(xor, message.getId());
                    localContext.putMessageIntoLocalTimeOfArrivalAndUpdateXor(message.getId());
                }
                log.trace("now my real xor = {}", localContext.getLocalTimeAndXor().getSecond());
                localContext.getPostService().postMessage(message);
                hisSize--;
                log.trace("got {}", message);
            }
        }

        messageQuery.close();
        log.info("synchronized successfully with {},  my xor ={}", socket, localContext.getLocalTimeAndXor().getSecond());
        log.debug("   xor = {}, theirXor = {}", xor, theirXor);
    }
}
